package com.itheima.article.stream;

import com.alibaba.fastjson.JSON;
import com.itheima.article.dto.ArticleVisitStreamMess;
import com.itheima.behaviour.dto.UpdateArticleMess;
import com.itheima.common.constants.BusinessConstants;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.Aggregator;
import org.apache.kafka.streams.kstream.Initializer;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

/**
 * @version 1.0
 * @description 说明
 * @package com.itheima.article.stream
 */
@Configuration
public class HotArticleStreamHandler {

    /**
     * 声明流处理。对文章行为的聚合
     * 在统计周期内，如果有多个行为操作针对同一篇时，把行为数量累计到一个消息
     * ArticleVisitStreamMess: like点赞数
     * zhangsan 1
     * lishi    1
     * wangwu   1
     * ArticleVisitStreamMess like=3 comment=2 collection=1
     *
     * @param streamsBuilder
     * @return
     */
    @Bean
    public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
        // 监听生产者的主题
        KStream<String, String> stream = streamsBuilder.stream(BusinessConstants.MqConstants.HOT_ARTICLE_SCORE_TOPIC);
        // 聚合统计
        stream.map((key,value)->{
            //value 就是消息内容 UpdateArticleMess JSON格式的字符串
            UpdateArticleMess msg = JSON.parseObject(value, UpdateArticleMess.class);
            Long articleId = msg.getArticleId();
            return new KeyValue<>(articleId.toString(), value);
        })
                //数据格式Map<articleId, UpdateArticleMessJSON>
                .groupBy((key,value)->key)
                // 处理10秒内的消息, 10秒一个周期
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))

                .aggregate(new Initializer<String>() {
                    // 初始化只执行一个,每个分组
                    @Override
                    public String apply() {
                        // 初始化时，同一个文章里面的行为数量都为0
                        ArticleVisitStreamMess articleVisitStreamMess = new ArticleVisitStreamMess();
                        return JSON.toJSONString(articleVisitStreamMess);
                    }
                }, new Aggregator<String, String, String>() {
                    @Override
                    //key=articleId, value=UpdateArticleMess JSON字符串,
                    //aggregate 上一次计算后的结果，如果没有上一次，那就是初始值 apply()的返回值
                    // 来一个消息就执行一次
                    public String apply(String key, String value, String aggregate) {
                        //第一次计算
                        //1. 解析上次的结果
                        ArticleVisitStreamMess lastResult = JSON.parseObject(aggregate, ArticleVisitStreamMess.class);
                        //2. 解析这次消息内容
                        UpdateArticleMess msg = JSON.parseObject(value, UpdateArticleMess.class);
                        lastResult.setArticleId(msg.getArticleId());
                        //3. 消息内容中操作的哪个行为
                        switch (msg.getType().getValue()) {
                            //4. 累计行为的值
                            //收藏
                            case 1: lastResult.setCollect(lastResult.getCollect() + msg.getNum());break;
                            //COMMENT 评论
                            case 2: lastResult.setComment(lastResult.getComment() + msg.getNum());break;
                            //LIKES
                            case 3: lastResult.setLike(lastResult.getLike() + msg.getNum());break;
                            //VIEWS
                            case 4: lastResult.setView(lastResult.getView() + msg.getNum());break;
                        }
                        //5. 转成json返回 lastResult给下个消息累计
                        System.out.println("===========>" + lastResult);
                        return JSON.toJSONString(lastResult);
                    }
                    // 给流处理一名称，唯一即可
                }, Materialized.as("hot-article-compute"))
                // 类型转换TimeWindow
                .toStream().map((key,value)->new KeyValue<>(key.key(), value))
                // 发给输出topic，由消费者监听
                .to(BusinessConstants.MqConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);
        return stream;
    }
}
